package org.redis.cluster.interceptor.support;

import java.util.Collection;
import java.util.HashMap;
import java.util.Map;

import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.collections.MapUtils;
import org.redis.cluster.RedisCluster;
import org.redis.cluster.extension.ExtensionLoader;
import org.redis.cluster.interceptor.Interceptor;
import org.redis.cluster.interceptor.Invocation;
import org.redis.cluster.plugin.Plugin;
import org.redis.cluster.processor.CommandProcessor;
import org.redis.cluster.redis.RedisCommand;
import org.redis.cluster.redis.transport.RedisReply;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class CommandProcessInterceptor implements Interceptor,Plugin {
	
	protected final Logger logger=LoggerFactory.getLogger(getClass());

	private RedisCluster redisCluster;
	private ExtensionLoader extensionLoader;
	
	private Map<String,CommandProcessor> processorMapping;
	
	@Override
	public void init(RedisCluster redisCluster) {
		this.redisCluster=redisCluster;
		this.extensionLoader=redisCluster.getExtensionLoader();
		this.processorMapping=initProcessorMapping();
	}
	
	@Override
	public void destroy() {
	}
	
	@Override
	public RedisReply intercept(Invocation invocation) throws Throwable {
		RedisCommand command=invocation.getCommand();
		CommandProcessor processor=getProcessor(command);
		if(processor==null) {
			return invocation.invoke();
		}
		
		if(logger.isDebugEnabled()) {
			logger.debug("处理 - {} -> {}",command,processor);
		}
		
		RedisReply reply=processor.process(command,invocation.getSession());
		if(reply!=null) {
			return reply;
		}
		
		return invocation.invoke();
	}
	
	protected CommandProcessor getProcessor(RedisCommand command) {
		return processorMapping.get(command.getCommand());
	}
	
	protected Map<String,CommandProcessor> initProcessorMapping() {
		Collection<CommandProcessor> processors=getProcessors();
		if(CollectionUtils.isEmpty(processors)) {
			return MapUtils.EMPTY_MAP;
		}
		
		Map<String,CommandProcessor> result=new HashMap<String,CommandProcessor>();
		for(CommandProcessor processor:processors) {
			result.put(processor.getSupportCommand().toUpperCase(), processor);
		}
		
		return result;
	}
	
	protected Collection<CommandProcessor> getProcessors() {
		return extensionLoader.load(CommandProcessor.class);
	}
}